package internal

import (
	"time"

	rabbitmq "github.com/rabbitmq/amqp091-go"
)

type RabbitMQ struct {
	conn *rabbitmq.Connection
	ch   *rabbitmq.Channel
	q    rabbitmq.Queue

	exchangeName string
	exchangeType string
	queueName    string
	bindingKey   string
}

type RabbitMQConfig struct {
	HostName string

	Username string
	Password string

	Vhost string
}

func NewRabbitMQ(config *RabbitMQConfig) (*RabbitMQ, error) {
	// 构建连接URL
	connURL := "amqp://" + config.Username + ":" + config.Password + "@" + config.HostName + "/" + config.Vhost

	// 建立连接
	conn, err := rabbitmq.Dial(connURL)
	if err != nil {
		return nil, err
	}

	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}

	return &RabbitMQ{
		conn: conn,
		ch:   ch,
	}, nil
}

// 声明Exchange
func (r *RabbitMQ) ExchangeDeclare(exchangeName, exchangeType string) error {
	r.exchangeName = exchangeName
	r.exchangeType = exchangeType

	return r.ch.ExchangeDeclare(
		exchangeName, // name
		exchangeType, // type
		true,         // durable
		false,        // auto-deleted
		false,        // internal
		false,        // no-wait
		nil,          // arguments
	)
}

// 声明Queue, 并绑定到Exchange
func (r *RabbitMQ) QueueDeclareAndBind(queueName string, bindingKey string) error {
	r.queueName = queueName
	r.bindingKey = bindingKey

	// 声明Queue
	err := r.QueueDeclare(queueName)
	if err != nil {
		return err
	}

	// 绑定Queue到Exchange
	return r.ch.QueueBind(
		r.q.Name,       // queue name
		bindingKey,     // routing key
		r.exchangeName, // exchange
		false,
		nil,
	)
}

func (r *RabbitMQ) QueueDeclare(queueName string) error {
	q, err := r.ch.QueueDeclare(
		queueName, // name
		true,      // durable
		false,     // delete when unused
		false,     // exclusive
		false,     // no-wait
		nil,       // arguments
	)
	if err != nil {
		return err
	}
	r.q = q
	return nil
}

func (r *RabbitMQ) Publish(messageID string, body []byte) error {
	return r.ch.Publish(
		r.exchangeName, // exchange
		r.bindingKey,   // routing key
		true,           // mandatory
		false,          // immediate
		rabbitmq.Publishing{
			Headers:         rabbitmq.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			DeliveryMode:    rabbitmq.Transient, // or amqp091.Persistent
			Priority:        0,                  // 0-9
			CorrelationId:   "",
			ReplyTo:         "",
			Expiration:      "",
			MessageId:       messageID,
			Timestamp:       time.Now(),
			Type:            "",
			AppId:           "",
			Body:            []byte(body),
		},
	)
}

func (r *RabbitMQ) Consume(consumerName string) (<-chan rabbitmq.Delivery, error) {
	return r.ch.Consume(
		r.q.Name,     // queue
		consumerName, // consumer
		false,        // auto-ack
		false,        // exclusive
		false,        // no-local
		false,        // no-wait
		nil,          // args
	)
}

func (r *RabbitMQ) Close() {
	if r.ch != nil {
		r.ch.Close()
	}

	if r.conn != nil {
		r.conn.Close()
	}
}
